// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#pragma once
#include <assert.h>
#include <atomic>
#include <functional>
#include <iostream>
#include <mutex>

namespace rocksdb {
// double linkd list that support concurrent operation, using std::atomic CAS
// function. need node compare function, to sort list in order.
template <typename T> class ConcurrentDLink {
private:
  std::atomic<T> head_;
  std::atomic<T> tail_;
  std::mutex mtx_;

public:
  ConcurrentDLink() {
    head_.store(nullptr);
    tail_.store(nullptr);
  }

  ConcurrentDLink(const ConcurrentDLink &) = delete;
  ConcurrentDLink(ConcurrentDLink &&t) = delete;
  ConcurrentDLink &operator=(const ConcurrentDLink &) = delete;
  ConcurrentDLink &operator=(ConcurrentDLink &&) = delete;

  ~ConcurrentDLink() {
    T cur = this->head_.load();
    T next = nullptr;
    while (cur != nullptr && (void *)cur != (void *)(this->tail_)) {
      next = cur->Next();
      delete cur;
      cur = next;
    }
  }

  T getHead() { return head_.load(); }
  T getTail() { return tail_.load(); }

  void insert(T newT, T next, std::function<int(T)> compp) {
    T headstart = nullptr, endStart = nullptr;
    // to make sure that value has not changed in concurrent access scene.
    if (tail_.compare_exchange_strong(endStart, newT,
                                      std::memory_order_relaxed)) {
      head_.compare_exchange_strong(headstart, newT, std::memory_order_relaxed);
      return;
    }

    if (next == nullptr) {
      next = tail_.load();
    }

    T prev = nullptr;
    bool needCmpNext = true;
    while (true) {
      int comp = 0;
      if (compp != nullptr)
        comp = compp(next);
      else
        comp = newT->CompareTo(next);

      assert(comp != 0);
      if (comp < 0) {               // while keys in next < newT
        if (next == tail_.load()) { // add node at list tail.
          if (!next->CASNext(nullptr, newT)) {
            continue;
          }
          tail_.compare_exchange_strong(next, newT);
          bool ok = newT->CASPrev(nullptr, next);
          assert(ok);

          return;
        }

        if (next->Next() == nullptr) {
          continue;
        }
        prev = next;
        next = next->Next();
        assert(next != prev);
        continue;
      }

      if (prev == nullptr) {
        if (next == head_.load()) { // add new node at list head.
          if (!next->CASPrev(nullptr, newT)) {
            continue;
          }
          bool ok = newT->CASNext(nullptr, next);
          assert(ok);
          head_.compare_exchange_strong(next, newT);
          return;
        }
        if (next->Prev() == nullptr) {
          continue;
        }

        prev = next->Prev();
        if (compp != nullptr)
          comp = compp(prev);
        else
          comp = newT->CompareTo(prev);

        assert(comp != 0);
        if (comp > 0) {
          next = prev;
          prev = nullptr;
          continue;
        } else {
          do {
            next = prev->Next();
          } while (next == nullptr);
          continue;
        }
      }

      // add new node between two nodes.
      if (!prev->CASNext(next, newT)) {
        prev = nullptr;
        continue;
      }
      newT->CASPrev(nullptr, prev);

      bool ok;
      do {
        ok = next->CASPrev(prev, newT);
      } while (!ok);

      newT->CASNext(nullptr, next);

      return;
    }
  }
};

} // namespace rocksdb